package com.gopay.scheduler.shedulerx;

import com.gopay.common.constants.scheduler.SchedulerConstants;
import com.gopay.common.util.JsonUtils;
import com.gopay.scheduler.GopayBaseJob;
import com.gopay.scheduler.shedulerx.alarm.AlarmEventBus;
import com.gopay.scheduler.shedulerx.event.JobEvent;
import com.gopay.scheduler.shedulerx.event.JobEventBus;
import com.gopay.scheduler.shedulerx.listener.GopaySchedulerListener;
import com.gopay.scheduler.shedulerx.remote.SchedulerHessianServiceExporter;
import com.gopay.scheduler.shedulerx.util.NetAddressUtils;
import com.gopay.scheduler.shedulerx.util.RedisOperator;
import com.gopay.scheduler.shedulerx.util.ThreadPoolUtil;
import com.gopay.scheduler.util.DateUtils;
import org.apache.commons.lang3.StringUtils;
import org.quartz.CronTrigger;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.Trigger;
import org.quartz.impl.JobDetailImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.quartz.MethodInvokingJobDetailFactoryBean;

import javax.sql.DataSource;
import java.io.IOException;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;


/**
 * @author 19371
 * 此类涉及到其他相关操作类的加载顺序，未避免NullPointException,请确保此类先加载,加载顺序请查看GopaySchedulerFactory的例子
 */
public class SchedulerOperateFactory {

    private final static Logger LOGGER = LoggerFactory.getLogger(SchedulerOperateFactory.class);

    private volatile static SchedulerOperateFactory eventFactory;
    /**
     * key: jobClass
     * value : eventControl
     */
    private final SchedulerHashMap<String,String> eventSet = new SchedulerHashMap<>();
    /**
     * key: jobName
     * value : cronExpression
     */
    private final SchedulerHashMap<String,String> jobCache = new SchedulerHashMap<>();
    /**
     * key:jobName
     * value : description
     */
    private final SchedulerHashMap<String,String> jobDescription = new SchedulerHashMap<>();
    /**
     * key:jobName
     * value: group
     */
    private final SchedulerHashMap<String,String> groupMapping = new SchedulerHashMap<>();
    /**
     * key : jobClass
     * value : jobName
     */
    private final SchedulerHashMap<String,String> eventMapping = new SchedulerHashMap<>();
    /**
     * key : jobName
     * value : jobClass
     */
    private final SchedulerHashMap<String,String> jobMapping = new SchedulerHashMap<>();
    /**
     * key: jobClass
     * value : level
     */
    private final SchedulerHashMap<String,String> alarmLevelMapping = new SchedulerHashMap<>();
    /**
     * key : jobName
     * value : count
     */
    private final SchedulerHashMap<String,Integer> alarmCountMapping = new SchedulerHashMap<>();
    /**
     * key:jobName
     * value: switchType
     */
    private final SchedulerHashMap<String, String> regionMapping = new SchedulerHashMap<>();
    /**
     * key:switchType
     * value:url
     */
    private final SchedulerHashMap<String, String> urlMapping = new SchedulerHashMap<>();
    /**
     * key:jobName
     * value:instance
     */
    private final SchedulerHashMap<String, GopayBaseJob> gopayBaseJobMapping = new SchedulerHashMap<>();
    private final List<String> regionList = new ArrayList<>();
    /**
     * 应用名字
     */
    private  String appName;
    /**
     * 事件Bus
     */
    private JobEventBus jobEventBus;
    /**
     * 报警Bus
     */
    private AlarmEventBus alarmEventBus;
    /**
     * 数据源
     */
    private DataSource dataSource;
    /**
     * 调度
     */
    private Scheduler scheduler;
    /**
     * 调度控制
     */
    private SchedulerControl schedulerControl;
    /**
     * spring上下文
     */
    private ApplicationContext applicationContext;
    /**
     * 运行环境
     */
    private String runEnv;

    private static final AtomicInteger REDIS_COUNTER = new AtomicInteger(0);

    private static final int FAIL_COUNT = 5;

    private static final Object LOCK_REDIS_PING = new Object();

    private volatile ScheduledThreadPoolExecutor redisPingExecutor = null;

    private volatile boolean schedulerX;

    private SchedulerOperateFactory() {
    }

    public static SchedulerOperateFactory getFactory() {
        if(eventFactory == null) {
            synchronized (SchedulerOperateFactory.class) {
                if (eventFactory == null) {
                    eventFactory = new SchedulerOperateFactory();
                }
            }
        }
        return eventFactory;
    }

    public HashMap<String, String> getJobCache() {
        return jobCache;
    }

    public HashMap<String, String> getGroupMapping() {
        return groupMapping;
    }

    private void setEventSet(HashMap<String,String> jobEvents){
        this.eventSet.putAll(jobEvents);
    }

    public HashMap<String, String> getEventSet() {
        return eventSet;
    }

    private void setAppName(String appName) {
        this.appName = appName;
    }

    public String getAppName() {
        return appName;
    }

    public SchedulerHashMap<String, String> getJobMapping() {
        return jobMapping;
    }

    private void setAlarmLevelMapping(HashMap<String, String> alarmLevelMap) {
        this.alarmLevelMapping.putAll(alarmLevelMap);
    }

    public HashMap<String, String> getAlarmLevelMapping() {
        return alarmLevelMapping;
    }


    public HashMap<String, Integer> getAlarmCountMapping() {
        return alarmCountMapping;
    }

    public List<String> getRegionList() {
        return regionList;
    }

    private void setAlarmCountMapping(HashMap<String, Integer> alarmCountMap) {
        this.alarmCountMapping.putAll(alarmCountMap);
    }

    public DataSource getDataSource() {
        return dataSource;
    }

    private void setDataSource(DataSource dataSource) {
        this.dataSource = dataSource;
    }

    private void setScheduler(Scheduler scheduler) {
        this.scheduler = scheduler;
    }

    public Scheduler getScheduler() {
        return scheduler;
    }

    public SchedulerControl getSchedulerControl() {
        return schedulerControl;
    }

    private void setSchedulerControl(SchedulerControl schedulerControl) {
        this.schedulerControl = schedulerControl;
    }

    public ApplicationContext getApplicationContext() {
        return applicationContext;
    }

    public HashMap<String, String> getJobDescription() {
        return jobDescription;
    }

    public String getRunEnv() {
        return runEnv;
    }

    private void setSpringEnv(ApplicationContext applicationContext) {
        this.applicationContext = applicationContext;
    }

    public boolean isSchedulerX() {
        return schedulerX;
    }

    public HashMap<String, String> getRegionMapping() {
        return regionMapping;
    }

    public SchedulerHashMap<String, String> getUrlMapping() {
        return urlMapping;
    }

    /**
     * 发布作业事件
     * @param event
     * @param className
     */
    public void postTraceEvent(JobEvent event, String className){
        event.setJobName(eventMapping.get(className));
        event.setAppName(appName+SchedulerConstants.SCHEDULER_HORIZONTAL_LINE+runEnv);
        event.setRegion(regionMapping.get(eventMapping.get(className)));
        event.setDecription(jobDescription.get(eventMapping.get(className)));
        jobEventBus.post(event);
    }

    /**
     * 发布misfire事件
     * @param event
     */
    public void postMisfireEvent(JobEvent event){
        event.setAppName(appName+SchedulerConstants.SCHEDULER_HORIZONTAL_LINE+runEnv);
        jobEventBus.post(event);
    }

    public void postSchedulerEvent(JobEvent event) {
        event.setAppName(appName+SchedulerConstants.SCHEDULER_HORIZONTAL_LINE+runEnv);
        jobEventBus.post(event);
    }

    /**
     * 发布报警事件
     * @param jobEvent
     * @param className
     */
    public void alarm(JobEvent jobEvent,String className) {
        jobEvent.setJobName(eventMapping.get(className));
        jobEvent.setAppName(appName+SchedulerConstants.SCHEDULER_HORIZONTAL_LINE+runEnv);
        jobEvent.setRegion(regionMapping.get(eventMapping.get(className)));
        jobEvent.setDecription(jobDescription.get(eventMapping.get(className)));
        alarmEventBus.alarm(jobEvent);
    }

    /**
     * 判断是否开启事件追踪
     * @param className
     * @return
     */
    public boolean isTrace(String className){
        String tracer = SchedulerOperateFactory.getFactory().getEventSet().get(className);
        return SchedulerConstants.EVENT_OPEN_TRACE.equals(tracer);
    }

    /**
     * 标记redis为等待运行
     * @param className
     */
    public void markWaiting(String className) {
        if(REDIS_COUNTER.get() > FAIL_COUNT){
            startPingRedis();
            return;
        }
        Map<String,Object> jsonMap = new HashMap<>(2);
        jsonMap.put(SchedulerConstants.REDIS_STATE_MARK,SchedulerConstants.JOB_WATING_STATE);
        jsonMap.put(SchedulerConstants.REDIS_UPDATE_TIME_MARK,DateUtils.format(new Date(),DateUtils.yyyy_MM_dd_HH_mm_ss));
        try {
            RedisOperator.setJobConfig(appName+SchedulerConstants.SCHEDULER_HORIZONTAL_LINE+SchedulerOperateFactory.getFactory().getRunEnv()+SchedulerConstants.SCHEDULER_HORIZONTAL_LINE+SchedulerConstants.SCHEDULER_MARK_CONFIG, eventMapping.get(className),JsonUtils.toJson(jsonMap));
            REDIS_COUNTER.set(0);
        } catch (Exception e) {
            exceptionLog(e,String.format("markRunning={classname=%s}",className));
            REDIS_COUNTER.incrementAndGet();
        }

    }

    /**
     * 标记redis为运行中
     * @param className
     * @param juid
     */
    public void markRunning(String className,String juid) {
        if(REDIS_COUNTER.get() > FAIL_COUNT){
            startPingRedis();
            return;
        }
        Map<String,Object> jsonMap = new HashMap<>(2);
        jsonMap.put(SchedulerConstants.REDIS_STATE_MARK,SchedulerConstants.JOB_RUNNING_STATE);
        jsonMap.put(SchedulerConstants.REDIS_UPDATE_TIME_MARK,DateUtils.format(new Date(),DateUtils.yyyy_MM_dd_HH_mm_ss));
        if(StringUtils.isNotBlank(juid)) {
            jsonMap.put(SchedulerConstants.REDIS_UPDATE_JUID_MARK, juid);
        }
        try {
            RedisOperator.setJobConfig(appName+SchedulerConstants.SCHEDULER_HORIZONTAL_LINE+SchedulerOperateFactory.getFactory().getRunEnv()+SchedulerConstants.SCHEDULER_HORIZONTAL_LINE+SchedulerConstants.SCHEDULER_MARK_CONFIG, eventMapping.get(className),JsonUtils.toJson(jsonMap));
            REDIS_COUNTER.set(0);
        } catch (Exception e) {
            exceptionLog(e,String.format("markRunning={classname=%s,juid=%s}",className,juid));
            REDIS_COUNTER.incrementAndGet();
        }
    }

    /**
     * 重置报警次数
     * @param className
     */
    public void resetAlarmCount(String className) {
        alarmCountMapping.put(eventMapping.get(className),0);
    }

    /**
     * 设置动态参数
     * @param jobName
     * @param luckyJson
     * @return
     * @throws IOException
     */
    public boolean setLuckyParam(String jobName, String luckyJson) throws IOException {
        String className = jobMapping.get(jobName);
        GopayBaseJob gopayBaseJob = gopayBaseJobMapping.get(className);
        gopayBaseJob.luckyParam(JsonUtils.jsonToMap(luckyJson));
        return true;
    }

    /**
     * 清除动态参数
     * @param jobName
     * @return
     */
    public boolean clearLuckyParam(String jobName){
        String className = jobMapping.get(jobName);
        GopayBaseJob gopayBaseJob = gopayBaseJobMapping.get(className);
        gopayBaseJob.clearParam();
        return true;
    }

    /**
     * 获取动态参数
     * @param jobName
     * @return
     * @throws IOException
     */
    public String getLuckyParam(String jobName) throws IOException {
        String className = jobMapping.get(jobName);
        GopayBaseJob gopayBaseJob = gopayBaseJobMapping.get(className);
        Map<String, Object> jobParam = gopayBaseJob.getParam();
        String luckyJson = "";
        if(jobParam != null && !jobParam.isEmpty()){
            luckyJson = JsonUtils.toJson(jobParam);
        }
        return luckyJson;
    }

    /**
     * 获取hessian url
     * @return
     */
    public String url(){
        SchedulerHessianServiceExporter export = applicationContext.getBean(SchedulerHessianServiceExporter.class);
        return NetAddressUtils.url()+applicationContext.getApplicationName()+SchedulerConstants.RPC_REMOTE+export.getName();
    }

    public  ParamBuilder builder() {
        return new ParamBuilder();
    }

    /**
     * 初始化需要，只加载一次
     */
    public final class ParamBuilder{

        public ParamBuilder schedulerControl(Scheduler scheduler) {
            getFactory().setScheduler(scheduler);
            return this;
        }

        public ParamBuilder appName(String appName){
            getFactory().setAppName(appName);
            return this;
        }

        /**
         * key: jobClass
         * value : eventControl
         * @param jobEvents
         * @return
         */
        public ParamBuilder eventSet(HashMap<String,String> jobEvents){
            getFactory().setEventSet(jobEvents);
            return this;
        }

        /**
         * key : jobClass
         * value : level
         * @param alarmLevelMap
         * @return
         */
        public ParamBuilder alarmLevelMapping(HashMap<String, String> alarmLevelMap) {
            getFactory().setAlarmLevelMapping(alarmLevelMap);
            return this;
        }

        /**
         * @see jobCache
         * @see jobDescription
         * @param triggerList
         * @return
         */
        public ParamBuilder triggers(List<Trigger> triggerList) {
            CronTrigger cronTrigger;
            for(Trigger trigger:triggerList){
                String jobName =trigger.getJobKey().getName();
                cronTrigger = (CronTrigger) trigger;
                jobCache.put(jobName,cronTrigger.getCronExpression());
                jobDescription.put(jobName,trigger.getDescription());
            }
            return this;
        }

        /**
         * 设置报警映射关系
         * key : jobName
         * value : count
         * @param alarmCountMap
         * @return
         */
        public ParamBuilder alarmCountMapping(HashMap<String, Integer> alarmCountMap) {
            getFactory().setAlarmCountMapping(alarmCountMap);
            return this;
        }

        /**
         * @see urlMapping
         * @see regionList
         * @see regionMapping
         * @param appMap
         * @return
         */
        public ParamBuilder appMap(Map<String, List<Trigger>> appMap) {
            Set<Map.Entry<String, List<Trigger>>> entries = appMap.entrySet();
            String url = url();
            for (Map.Entry<String, List<Trigger>> entry : entries){
                List<Trigger> triggers = entry.getValue();
                for (Trigger trigger : triggers){
                    regionMapping.put(trigger.getJobKey().getName(),entry.getKey());
                }
                urlMapping.put(entry.getKey(),url);
                regionList.add(entry.getKey());
            }
            return this;
        }

        /**
         * spring环境
         * @see gopayBaseJobMapping
         * @see groupMapping
         * @see jobMapping
         * @see eventMapping
         * @param applicationContext
         * @return
         */
        public ParamBuilder springEnv(ApplicationContext applicationContext) {
            getFactory().setSpringEnv(applicationContext);
            getFactory().setDataSource(applicationContext.getBean(SchedulerConstants.SCHEDULER_DATA_SOURCE,DataSource.class));
            Set<Map.Entry<String, String>> entries = jobCache.entrySet();
            for (Map.Entry<String,String> entry : entries) {
                JobDetailImpl jobDetail = applicationContext.getBean(entry.getKey(), JobDetailImpl.class);
                MethodInvokingJobDetailFactoryBean methodInvoker = (MethodInvokingJobDetailFactoryBean) jobDetail.getJobDataMap().get("methodInvoker");
                String jobName = jobDetail.getKey().getName();
                String group = jobDetail.getKey().getGroup();
                String className = methodInvoker.getTargetClass().getName();
                gopayBaseJobMapping.put(className,(GopayBaseJob) methodInvoker.getTargetObject());
                groupMapping.put(jobName,group);
                jobMapping.put(jobName,className);
                eventMapping.put(className,jobName);
            }
            return this;
        }

        /**
         * 运行环境
         * @param schedulerEnv
         * @return
         */
        public ParamBuilder runEnv(String schedulerEnv) {
            runEnv = schedulerEnv;
            return this;
        }

        /**
         * 结束
         * @return
         */
        public SchedulerOperateFactory ending(){
            return getFactory();
        }

    }

    /**
     * init config
     * @throws SchedulerException
     * @throws SQLException
     */
    public void initJob() throws Exception {
        long startTime = System.currentTimeMillis();
        do{
            LOGGER.info("<<<<<<<|| the scheduler config init start ||>>>>>>>");
            /**
             * init the event bus
             */
            jobEventBus = JobEventBus.getSingle();
            alarmEventBus = AlarmEventBus.getSingle();
            /**
             * init the job control
             */
            getFactory().setSchedulerControl(SchedulerControl.getSingle(scheduler));
            /**
             * init scheduler listener
             */
            GopaySchedulerListener.getSingle(scheduler.getListenerManager()).init();

        }/**
         *  init scheduler config
         */
        while (!SchedulerConfigLoad.getSingle().init());
        schedulerX = true;
        LOGGER.info(String.format("<<<<<<<|| the scheduler config init over {%sms} ||>>>>>>>",(System.currentTimeMillis()-startTime)));
    }

    class SchedulerHashMap<K,V> extends HashMap<K,V>{
        @Override
        public V remove(Object key) {
            throw new UnsupportedOperationException();
        }
    }


    class RedisPingTask implements Runnable{
        private String key;

        public RedisPingTask(String key) {
            this.key = key;
        }

        @Override
        public void run() {
            LOGGER.info("-----keeping the redis-ping-check-------");
            try {
                RedisOperator.ping(appName+SchedulerConstants.SCHEDULER_HORIZONTAL_LINE+runEnv+SchedulerConstants.SCHEDULER_HORIZONTAL_LINE+SchedulerConstants.REDIS_PING
                        ,key,"ping",2);
                REDIS_COUNTER.set(0);
            } catch (Exception e) {
                exceptionLog(null,"-----SchedulerX RedisPingTask error，please check redis's health-----");
            }
        }
    }

    private void startPingRedis() {
        if(redisPingExecutor == null) {
            synchronized (LOCK_REDIS_PING) {
                if(redisPingExecutor == null) {
                    redisPingExecutor = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {
                        int index = 0;

                        @Override
                        public Thread newThread(Runnable runnable) {
                            ++this.index;
                            return new Thread(runnable, "redis-ping-thread-" + this.index);
                        }
                    });
                    String key = String.valueOf(new Random().nextInt());
                    redisPingExecutor.scheduleAtFixedRate(new RedisPingTask(key),5,5,TimeUnit.SECONDS);
                }
            }
        }

    }

    public void shutdown(){
        try {
            if(scheduler !=null && !scheduler.isShutdown()) {
                scheduler.shutdown();
            }
        } catch (SchedulerException e) {
            exceptionLog(e,"-----------------《scheduler shutdown fail》----------------------");
        }
        JobEventBus.getSingle().shutdown();
        AlarmEventBus.getSingle().shutdown();
        SchedulerConfigLoad.getSingle().shutdown();
        if(redisPingExecutor != null) {
            String poolName = String.format("from class {%s} , and poolName is {%s}",this.getClass().getName(),"REDIS PING 线程池");
            ThreadPoolUtil.shutdownGracefully(poolName,redisPingExecutor,TimeUnit.MILLISECONDS,10);
        }
    }


    private static void exceptionLog(Exception e,String description){
        LOGGER.error(String.format("the scheduler occur error ,the description is {%s}, e={%s}",description,e == null ? "redis异常" : e.getMessage()));
    }
}
